package com.cictec.middleware.commons.utils.elasticsearch;

import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
import org.elasticsearch.action.admin.indices.exists.types.TypesExistsRequest;
import org.elasticsearch.action.admin.indices.exists.types.TypesExistsResponse;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.cictec.middleware.commons.utils.DateUtils;
import com.cictec.middleware.commons.utils.PropertiesUtils;
import com.fasterxml.jackson.core.JsonParseException;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonMappingException;

/**
 * elasticsearch 相关操作工具类
 * 
 * <br> 【数据对应关系】
 * <br> 	index(索引)    --> 数据库名
 * <br> 	type(类型)     --> 表名
 * <br> 	document(文档) --> 表中一行数据
 * 
 * @author Administrator
 *
 */
//@SuppressWarnings({ "rawtypes", "unchecked" })
public class ESClientUtils {

	private static Logger logger = LoggerFactory.getLogger(ESClientUtils.class);
	
	private static Client client;
	
	/**
	 * 获得连接
	 * 
	 * @return
	 * @throws UnknownHostException
	 */
	public static Client getClient() throws UnknownHostException {
		String host = PropertiesUtils.getString("elasticsearch.ip");		//  es服务器的host
		int port = PropertiesUtils.getIntValue("elasticsearch.port", 9300); // es服务器暴露给client的port
		String clusterName = PropertiesUtils.getString("elasticsearch.cluster.name");	// 集群名称
		
//		String host = "127.0.0.1";		//  es服务器的host
//		int port = 9300; // es服务器暴露给client的port
//		String clusterName = "cictec";	// 集群名称

		
		
		
//		Client client = TransportClient.builder().build().addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), port)); // 单机环境
		
		Settings settings = Settings.settingsBuilder().put("cluster.name", clusterName).build();		// 设置集群名称
		Client client = TransportClient.builder().settings(settings).build().addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), port));
		return client;

	}
	
	/**
	 * 批量新增
	 * @param eSModelList		es存储模型的列表(具体看ESModel)
	 * @throws UnknownHostException
	 * @throws JsonProcessingException
	 */
	public static void addDocumentList(List<ESModel> eSModelList) throws UnknownHostException, JsonProcessingException {
		
		if(eSModelList == null || eSModelList.size() == 0){
			return;
		}
		
		
		Client client = getClient();

		// 批量处理request
		BulkRequestBuilder bulkRequest = client.prepareBulk();

		for (ESModel model : eSModelList) {
			bulkRequest.add(new IndexRequest(model.getIndex(), model.getType(), model.getId()).source(model.getJsonString()));
		}

		// 执行批量处理request
		BulkResponse bulkResponse = bulkRequest.get();

		// 处理错误信息
		if (bulkResponse.hasFailures()) {
			logger.error("====================批量创建索引过程中出现错误 下面是错误信息==========================");
			System.out.println("====================批量创建索引过程中出现错误 下面是错误信息==========================");
			long count = 0L;
			for (BulkItemResponse bulkItemResponse : bulkResponse) {
				System.out.println("发生错误的 索引id为 : " + bulkItemResponse.getId() + " ，错误信息为：" + bulkItemResponse.getFailureMessage());
				logger.error("发生错误的 索引id为 : " + bulkItemResponse.getId() + " ，错误信息为：" + bulkItemResponse.getFailureMessage());
				count++;
			}
			System.out.println( "====================批量创建索引过程中出现错误 上面是错误信息 共有: " + count + " 条记录==========================");
			logger.error("====================批量创建索引过程中出现错误 上面是错误信息 共有: " + count + " 条记录==========================");
		}

		client.close();
	}
	
	/**
	 * 新增document
	 * 
	 * @param index
	 *            索引名称
	 * @param type
	 *            类型名称
	 * @param goods
	 *            存储模型对象
	 * @throws UnknownHostException
	 */
	public static void addDocument(String index, String type, ESModel model) throws UnknownHostException {
		
		if(client == null){
			client = getClient();
		}

		client.prepareIndex(index, type, model.getId()).setSource(model.getJsonString()).get();

		// 单条插入 不关闭连接
//		client.close();
	}

	/**
	 * 新增document
	 * @param model
	 * @throws UnknownHostException
	 */
	public static void addDocument(ESModel model) throws UnknownHostException {
		
		if(client == null){
			client = getClient();
		}

		client.prepareIndex(model.getIndex(), model.getType(), model.getId()).setSource(model.getJsonString()).get();

		// 单条插入 不关闭连接
//		client.close();
	}

	/**
	 * 删除document
	 * 
	 * @param index
	 *            索引名称
	 * @param type
	 *            类型名称
	 * @param modelId
	 *            要删除存储模型对象的id
	 * @throws UnknownHostException
	 */
	public static void deleteDocument(String index, String type, String modelId) throws UnknownHostException {
//		Client client = getClient();
		if(client == null){
			client = getClient();
		}

		client.prepareDelete(index, type, modelId + "").get();
//		client.close();
	}
	
	public static void deleteIndex(ESModel esModel) throws UnknownHostException {
//		Client client = getClient();
		if(client == null){
			client = getClient();
		}
		
		client.prepareDelete(esModel.getIndex(), esModel.getType(), esModel.getId()).get();
		
//		client.close();
	}
	
	/**
	 * 批量刪除索引
	 * @param listVedio
	 * @return
	 */
	public static int deleteIndex(List<ESModel> eSModelList) throws UnknownHostException {
		
		Client client = getClient();

		BulkRequestBuilder deleteBulk = client.prepareBulk();
		for (ESModel esModel : eSModelList) {
			deleteBulk.add(client.prepareDelete(esModel.getIndex(), esModel.getType(), esModel.getId()));
		}
		deleteBulk.execute().actionGet();
		
		client.close();
		return 1;
	}
	

	/**
	 * 更新document
	 * 
	 * @param index 索引名称
	 * @param type  类型名称
	 * @param goods
	 *            商品dto
	 * @throws JsonProcessingException
	 * @throws UnknownHostException
	 */
	public static void updateDocument(String index, String type, ESModel model)
			throws UnknownHostException, JsonProcessingException {
		// 如果新增的时候index存在，就是更新操作
		addDocument(index, type, model);
	}
	
	/**
	 * 判断指定的索引名是否存在
	 * 
	 * @param indexName 索引名
	 * @return 存在：true; 不存在：false;
	 */
	public static boolean isExistsIndex(String indexName) throws Exception {
		if(client == null){
			client = getClient();
		}
		
		IndicesExistsResponse  response = client.admin().indices().exists(new IndicesExistsRequest().indices(new String[]{indexName})).actionGet();
        return response.isExists();
	}

	/**
	 * 判断指定的索引的类型是否存在
	 * 
	 * @param indexName 索引名
	 * @param indexType 索引类型
	 * @return 存在：true; 不存在：false;
	 */
	public static boolean isExistsType(String indexName, String indexType) throws Exception {
		if(client == null){
			client = getClient();
		}
		
		if(!isExistsIndex(indexName)){
			return false;
		}
		
		TypesExistsResponse response = client.admin().indices().typesExists(new TypesExistsRequest(new String[] { indexName }, indexType)).actionGet();
		return response.isExists();
	}
	
	
	/**
	 * 删除es中位置信息
	 * @param date
	 * @param busUuid
	 * @param busNumber
	 * @param lineUuid
	 * @throws JsonParseException
	 * @throws IOException
	 */
	public static void deletePostion(String date, String busUuid, String busNumber, String lineUuid) throws Exception {
		// 执行2次 应该能删完 一辆车 一天不会超过20000条
		List<ESModel> esModelList = searchPostion(date, busUuid, busNumber, lineUuid);	// 默认是10000条数据
		if(esModelList.size() > 0){
			deleteIndex(esModelList);

			if(esModelList.size() == 10000){
				esModelList = searchPostion(date, busUuid, busNumber, lineUuid);	// 默认是10000条数据
				if(esModelList.size() > 0){
					deleteIndex(esModelList);
				}
			}
		}
	}
	/**
	 * 删除es中位置信息
	 * @param date
	 * @param busUuid
	 * @param busNumber
	 * @throws Exception
	 */
	public static void deletePostion(String date, String busUuid, String busNumber) throws Exception {
		// 执行2次 应该能删完 一辆车 一天不会超过20000条
		List<ESModel> esModelList = searchPostion(date, busUuid, busNumber);	// 默认是10000条数据
		if(esModelList.size() > 0){
			deleteIndex(esModelList);
			
			if(esModelList.size() == 10000){
				esModelList = searchPostion(date, busUuid, busNumber);	// 默认是10000条数据
				if(esModelList.size() > 0){
					deleteIndex(esModelList);
				}
			}
		}
	}
	
	public static List<ESModel> searchPostion(String date, String busUuid, String busNumber) throws Exception {
		return searchPostion(date, busUuid, busNumber, null);
	}

	/**
	 * 查询位置信息
	 * @param date
	 * @param busUuid
	 * @param busNumber
	 * @param lineUuid
	 * @return
	 * @throws JsonParseException
	 * @throws IOException
	 */
	public static List<ESModel> searchPostion(String date, String busUuid, String busNumber, String lineUuid) throws Exception {
		String tableIndex = PropertiesUtils.getString("elasticsearch.index.position");
		String tableType = busNumber;

		if(!ESClientUtils.isExistsType(tableIndex, tableType)){
			return new ArrayList<ESModel>();
		}
		
		
		Client client = getClient();
		
		Date startTime = DateUtils.getStartDateTime(date);
		Date endTime = DateUtils.getEndDateTime(date);
		

		QueryBuilder queryBuilder;
		if(lineUuid == null){
			queryBuilder = QueryBuilders.boolQuery()
//					.must(QueryBuilders.termQuery("date", date))
					.must(QueryBuilders.termQuery("busUuid", busUuid))
//					.must(QueryBuilders.termQuery("lineUuid", lineUuid))
					.must(QueryBuilders.rangeQuery("gatherTime").from(startTime.getTime()).to(endTime.getTime()))
					;
		}else{
			queryBuilder = QueryBuilders.boolQuery()
//					.must(QueryBuilders.termQuery("date", date))
					.must(QueryBuilders.termQuery("busUuid", busUuid))
					.must(QueryBuilders.termQuery("lineUuid", lineUuid))
					.must(QueryBuilders.rangeQuery("gatherTime").from(startTime.getTime()).to(endTime.getTime()))
					;
		}
		
		
		SearchResponse response = client.prepareSearch(tableIndex)
				.setTypes(tableType)
				.setSearchType(SearchType.DFS_QUERY_THEN_FETCH)
				.setQuery(queryBuilder)
				.setSize(10000)
				.execute()
				.actionGet();
		//获取命中数
//		System.out.println(response.getHits().totalHits());
		//获取响应字符串
//		System.out.println(response.toString());
		//遍历查询结果输出相关度分值和文档内容
		SearchHits searchHits =  response.getHits();
		
		List<ESModel> dataList = new ArrayList<>();
		ESModel model = null;
		for(SearchHit searchHit : searchHits){
//	        System.out.println(searchHit.getScore());
//	        System.out.println(searchHit.getSourceAsString());
			String json = searchHit.getSourceAsString();
			
			model = new ESModel();
			model.setId(searchHit.getId());
			model.setIndex(tableIndex);
			model.setType(tableType);
			model.setJsonString(json);
			
			dataList.add(model);
		}
		
		client.close();
		
		return dataList;
	}
	
	/**
	 * 删除指定天的进出站信息
	 * @param date
	 * @param plate
	 * @throws Exception
	 */
	public static void deleteEnterOut(String date, String plate) throws Exception{
		
		List<ESModel> oldEnterList = ESClientUtils.searchEnterOut(date, plate);	// 默认是10000条数据
		if(oldEnterList.size() > 0){
			logger.debug("删除es进出站信息：{} {}",date, plate);
			ESClientUtils.deleteIndex(oldEnterList);
		}
	}
	
	/**
	 * 查询进出站数据
	 * @param date
	 * @param busNumber
	 * @return
	 * @throws Exception
	 */
	public static List<ESModel> searchEnterOut(String date, String busNumber) throws Exception {
		
		return searchEnterOut(date, busNumber, null);
	}

	/**
	 * 查询进出站数据
	 * @param date
	 * @param busNumber
	 * @param lineUuid
	 * @return
	 * @throws JsonParseException
	 * @throws JsonMappingException
	 * @throws IOException
	 */
	public static List<ESModel> searchEnterOut(String date, String busNumber, String lineUuid) throws Exception {
		String tableIndex = PropertiesUtils.getString("elasticsearch.index.enterout");
//		String tableIndex = "bus-enteroutstation";
		String tableType = busNumber + "-enterOut";

		Client client = getClient();

		if(!ESClientUtils.isExistsType(tableIndex, tableType)){
			return new ArrayList<ESModel>();
		}
		
		
//		Date startTime = DateUtils.getStartDateTime(date);
//		Date endTime = DateUtils.getEndDateTime(date);
		
		QueryBuilder queryBuilder;
		if(lineUuid == null){
			queryBuilder = QueryBuilders.boolQuery()
					.must(QueryBuilders.termQuery("date", date))
//					.must(QueryBuilders.termQuery("line_uuid", lineUuid))
//					.must(QueryBuilders.termQuery("bus_plate_number", busNumber))// 不能拿车牌号，因为有分词
//					.must(QueryBuilders.rangeQuery("enter_time").from(startTime.getTime()).to(endTime.getTime()))
					;
		}else{
			queryBuilder = QueryBuilders.boolQuery()
					.must(QueryBuilders.termQuery("date", date))
					.must(QueryBuilders.termQuery("line_uuid", lineUuid))
//					.must(QueryBuilders.termQuery("bus_plate_number", busNumber))// 不能拿车牌号，因为有分词
//					.must(QueryBuilders.rangeQuery("enter_time").from(startTime.getTime()).to(endTime.getTime()))
					;
		}
		
		
		SearchResponse response = client.prepareSearch(tableIndex)
				.setTypes(tableType)
				.setSearchType(SearchType.DFS_QUERY_THEN_FETCH)
				.setQuery(queryBuilder)
				.setSize(10000)
				.execute()
				.actionGet();
		//获取命中数
//		System.out.println(response.getHits().totalHits());
		//获取响应字符串
//		System.out.println(response.toString());
		//遍历查询结果输出相关度分值和文档内容
		SearchHits searchHits =  response.getHits();
		
		List<ESModel> dataList = new ArrayList<>();
		ESModel model = null;
		for(SearchHit searchHit : searchHits){
//	        System.out.println(searchHit.getScore());
//	        System.out.println(searchHit.getSourceAsString());
			String json = searchHit.getSourceAsString();
			
			model = new ESModel();
			model.setId(searchHit.getId());
			model.setIndex(tableIndex);
			model.setType(tableType);
			model.setJsonString(json);
			
			dataList.add(model);
		}
		
		client.close();
		return dataList;
	}
	
	public static void main(String ...strings) throws Exception{
//		Date startDate = DateUtils.getStartDateTime("2016-08-16");
//		Date endDate = DateUtils.getEndDateTime("2016-08-16");
/*		
		List<ESModel> list = searchPostion("2016-08-16", "761467e41d7d41c88636", "陕J54836", "174572ad4e0d493aaa38");
//		List<ESModel> list = searchEnterOut("陕J54863", "174572ad4e0d493aaa38", startDate, endDate);
		for (ESModel esModel : list) {
			System.out.println(esModel.getId() + "\t" + esModel.getJsonString());
		}
		*/
//		deletePostion("2016-08-16", "577000039f634bdb9328", "陕J54863", "174572ad4e0d493aaa38");
//		searchEnterOut("2016-08-16", "ss", "ff");

//		System.out.println(isExistsIndex("bus-enteroutstation"));
//		System.out.println(isExistsType("bus-enteroutstsation", "粤J42771"));
		
//		List list = searchEnterOut("2016-08-16", "陕J54863", "174572ad4e0d493aaa38");
//		System.out.println(list.size());
	}

}